package minicloud;

import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DriverManagerDataSource;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;

@Slf4j
public class MysqlPersistence {
    public final JdbcTemplate jdbcTemplate;
    public static final DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:00");
    public final String taskId; // 当前任务的唯一标识

    // 初始化MySQL连接

    /**
     * setx MYSQL_HOST  mysql.com /M
     * setx MYSQL_PORT  3306 /M
     * setx MYSQL_USER  root /M
     * setx MYSQL_PWD  root /M
     *


     MAC env:

     vi ~/.zshrc
     # MySQL 环境变量
     export MYSQL_HOST=mysql.com
     export MYSQL_PORT=3306
     export MYSQL_USER=root
     export MYSQL_PWD=root

     source ~/.zshrc


     *
     */
    public MysqlPersistence(String taskId) {
        this.taskId = taskId;
        String host = System.getenv("MYSQL_HOST");
        String port = System.getenv("MYSQL_PORT");
        String user = System.getenv("MYSQL_USER");
        String pwd = System.getenv("MYSQL_PWD");
        log.info("Mysql: " + host + ":" + port);
        log.info("Mysql: " + user + ":" + pwd);

        DriverManagerDataSource dataSource = new DriverManagerDataSource();
        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
        dataSource.setUrl("jdbc:mysql://" + host + ":3306/stream_db?useSSL=false&serverTimezone=UTC");
        dataSource.setUsername(user);
        dataSource.setPassword(pwd); // 替换为你的MySQL密码
        this.jdbcTemplate = new JdbcTemplate(dataSource);
    }

    // 1. 读取指定任务的未处理日志（增加task_id过滤）
    public List<LogEntry> readUnprocessedLogs(long lastId, int batchSize) {
        return jdbcTemplate.query(
                "SELECT id, level, message, create_time FROM raw_logs " +
                        "WHERE task_id = ? AND id > ? ORDER BY id LIMIT ?",
                (rs, row) -> new LogEntry(
                        rs.getLong("id"),
                        rs.getString("level"),
                        rs.getString("message"),
                        rs.getTimestamp("create_time").toLocalDateTime(),
                        taskId // 日志绑定到当前任务
                ),
                taskId, lastId, batchSize
        );
    }

    // 2. 保存指定任务的统计结果
    public void saveStatistics(LocalDateTime windowStart, int totalCount) {
        jdbcTemplate.update(
                "INSERT INTO error_statistics (task_id, window_start_time, error_count) " +
                        "VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE error_count = ?",
                taskId, windowStart, totalCount, totalCount
        );
    }

    // 3. 读取指定任务的状态（增加task_id过滤）
    public String getState(String key) {
        // 初始化状态：如果任务的状态不存在，自动创建默认值
        Integer count = jdbcTemplate.queryForObject(
                "SELECT COUNT(*) FROM processing_state WHERE task_id = ? AND state_key = ?",
                Integer.class, taskId, key
        );
        if (count == 0) {
            String defaultValue = key.equals("last_processed_id") ? "0" :
                    (key.equals("current_window_count") ? "0" : "");
            jdbcTemplate.update(
                    "INSERT INTO processing_state (task_id, state_key, state_value) VALUES (?, ?, ?)",
                    taskId, key, defaultValue
            );
            return defaultValue;
        }
        return jdbcTemplate.queryForObject(
                "SELECT state_value FROM processing_state WHERE task_id = ? AND state_key = ?",
                String.class, taskId, key
        );
    }

    // 4. 更新指定任务的状态
    public void updateState(String key, String value) {
        jdbcTemplate.update(
                "UPDATE processing_state SET state_value = ? WHERE task_id = ? AND state_key = ?",
                value, taskId, key
        );
    }

    // 辅助方法：获取当前窗口开始时间（如 10:00:00）
    public LocalDateTime getCurrentWindowStart() {
        return LocalDateTime.now().format(dtf).equals(getState("current_window_start"))
                ? LocalDateTime.parse(getState("current_window_start"), dtf)
                : LocalDateTime.now().withSecond(0).withNano(0);
    }
}
